package com.zyx.pulsardemo.functions.sdk;


import org.apache.commons.collections4.queue.CircularFifoQueue;
import org.apache.pulsar.functions.api.Context;
import org.apache.pulsar.functions.api.Function;

import java.io.*;
import java.nio.ByteBuffer;

/**
 * @author Yaxi.Zhang
 * @since 2021/10/8 18:14
 * desc: 读取传感器平均值
 */
public class AvgSensorReadingFunction implements Function<Double, Void> {

    private static final String VALUES_KEY = "VALUES";

    @Override
    public Void process(Double input, Context ctx) throws Exception {

        // 我们对用于存储以前的传感器读数 的Java对象进行反序列化
        CircularFifoQueue<Double> values = getValues(ctx);

        if (Math.abs(input - getAverage(values)) > 10.0) {
            // trigger an alarm.
            // 如果当前读数显著不同, 则生成警报
        }
        // 将当前值添加到观察值列表中
        values.add(input);

        // 将更新后的Java对象存储在状态中
        ctx.putState(VALUES_KEY, serialize(values));

        return null;
    }

    private Double getAverage(CircularFifoQueue<Double> values) {
        // 使用Streams API来计算平均值
        return values.stream().mapToDouble(Double::doubleValue).sum() / values.size();
    }

    private CircularFifoQueue<Double> getValues(Context ctx) {
        if (ctx.getState(VALUES_KEY) == null) {
            // 如果状态存储中不存在Java对象, 则实例化该Java对象
            return new CircularFifoQueue<>(100);
        } else {
            // 将状态存储中的字节转换为我们需要的Java对象
            return deserialize(ctx.getState(VALUES_KEY));
        }
    }

    /**
     * 将 CircularFifoQueue 对象序列化为 二进制 流
     */
    private ByteBuffer serialize(CircularFifoQueue<Double> values) {
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        try (ObjectOutputStream out = new ObjectOutputStream(bos); ) {
            out.writeObject(values);
            out.flush();
            return ByteBuffer.wrap(bos.toByteArray());
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 将 二进制流 反序列化为 CircularFifoQueue 对象
     */
    @SuppressWarnings("unchecked")
    private CircularFifoQueue<Double> deserialize(ByteBuffer buffer) {

        try (ByteArrayInputStream bis = new ByteArrayInputStream(buffer.array());
             ObjectInput in = new ObjectInputStream(bis);) {
            return (CircularFifoQueue<Double>) in.readObject();

        } catch (IOException | ClassNotFoundException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        return null;
    }
}
